﻿using Gasgoo.ShareLibrary.Framework.iQueueProvider;
using Gasgoo.ShareLibrary.Framework.Options;
using Microsoft.Extensions.DependencyInjection;
using Newtonsoft.Json;
using Orleans.Runtime;
using System;
using System.Collections.Generic;
using System.Data;
using System.Data.SqlClient;
using System.Linq;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Threading.Tasks;

namespace Gasgoo.QueueProvider
{
    public class SQLServerQueueStorageProvider : IQueueStorageProvider
    {
        private Subject<BatchItemOptions<MessageDataOptions>> _subject;

        public string QueueId { get; set; }
        private string QueueTableName { get; set; }
        private string QueueOffsetTableName { get; set; }
        public string ProviderName { get; }

        private AdoNetOptions options;
        string connection;

        public SQLServerQueueStorageProvider(string providerName)
        {
            ProviderName = providerName;
            //this.client = provider.GetRequiredServiceByName<SqlSugarClient>(providerName);
        }

        //public SQLServerQueueStorageProvider(string providerName, IServiceProvider provider)
        //{
        //    this.client = provider.GetRequiredServiceByName<SqlSugarClient>(providerName);
        //}

        private void Ini(AdoNetOptions options, string queueId)
        {
            // Set Connection option
            {
                this.options = options;
                this.connection = this.options.GetConnection(this.options.QueueDatabaseName);
            }

            // 批处理器
            {
                _subject = new Subject<BatchItemOptions<MessageDataOptions>>();
                _subject.Buffer(TimeSpan.FromMilliseconds(10), 100) // 100毫秒 或者 100 条
                    .Where(x => x.Count > 0)
                    .Select(list => Observable.FromAsync(() => BatchWriteAsync(list)))
                    .Concat()
                    .Subscribe();
            }

            // 检查存储表创建
            {
                // 创建实例
                this.QueueId = queueId;
                this.QueueTableName = $"ClusterQueueTable_{this.QueueId}";
                this.QueueOffsetTableName = $"ClusterQueueOffsetTable_{this.QueueId}";
                // 下一次获取应该 > offset, offset 是最后一个消费的值
                // 创建表脚本
                string createTable = $@"CREATE TABLE {this.QueueTableName}
							(
								offset int identity(1,1) primary key not null,
								payload nvarchar(max)
							);
                            CREATE TABLE {this.QueueOffsetTableName}
							(
								id int identity(1,1) primary key not null,
								offset int default 0
							);
                            insert into {this.QueueOffsetTableName}(offset) values(0);
                            ";

                using (SqlConnection conn = new SqlConnection(this.connection))
                {
                    List<MessageDataOptions> messages = new List<MessageDataOptions>();
                    using (SqlCommand command = new SqlCommand($"select 1 from sysobjects where id=object_Id('{this.QueueTableName}')", conn))
                    {
                        conn.Open();
                        var result = command.ExecuteScalar();

                        if (result == null)
                        {
                            // 不存在创建表
                            command.CommandText = createTable;
                            command.ExecuteNonQuery();
                        }
                    }
                }


            }
        }

        public IQueueStorageProvider CreateInstance(IServiceProvider provider, string queueId)
        {
            queueId = queueId.Replace("-", "_");
            var queueStorageProvider = new SQLServerQueueStorageProvider(this.ProviderName);
            AdoNetOptions options = provider.GetRequiredService<AdoNetOptions>();  
            queueStorageProvider.Ini(options, queueId);
            return queueStorageProvider;
        }

        public async Task<IEnumerable<MessageDataOptions>> DequeueAsync(int maxCount)
        {

            // 这里应该自己维护消费信息
            await using (SqlConnection conn = new SqlConnection(this.connection))
            {
                List<MessageDataOptions> messages = new List<MessageDataOptions>();
                await using (SqlCommand command = new SqlCommand($"select top {maxCount} offset,payload from {this.QueueTableName} where offset > (select top 1 offset from {this.QueueOffsetTableName})", conn))
                {
                    await conn.OpenAsync();
                    var reader = await command.ExecuteReaderAsync();
                    if (reader != null)
                    {
                        int lastOffste = 0;
                        while (await reader.ReadAsync())
                        {
                            var message = reader.GetString(1); //  reader["payload"].ToString();
                            if (!string.IsNullOrWhiteSpace(message))
                            {
                                var options = GasgooUtils.DeserializeObject<MessageDataOptions>(message);
                                options.SequenceNumber = lastOffste = reader.GetInt32(0);
                                messages.Add(options);
                            }
                        }
                        await reader.CloseAsync();
                        if (lastOffste > 0)
                        {
                            // 更新offset
                            command.CommandText = $"update {this.QueueOffsetTableName} set offset = {lastOffste}";
                            command.ExecuteNonQuery();
                        }
                    }
                }

                //List<MessageDataOptions> messages = new List<MessageDataOptions>();
                //await using (SqlCommand command = new SqlCommand(, conn))
                //{
                //    command.Parameters.AddRange(new SqlParameter[]
                //    {
                //            new SqlParameter("startOffset",offset)
                //    });
                //    await conn.OpenAsync();
                //    var reader = await command.ExecuteReaderAsync();

                //    if (reader != null)
                //    {
                //        while (await reader.ReadAsync())
                //        {
                //            var message = reader.GetString(1); //  reader["payload"].ToString();
                //            if (!string.IsNullOrWhiteSpace(message))
                //            {
                //                var options = GasgooUtils.DeserializeObject<MessageDataOptions>(message);
                //                options.SequenceNumber = reader.GetInt32(0);
                //                messages.Add(options);
                //            }
                //        }
                //    }
                //}

                return messages;
            }

        }

        public async Task EnqueueAsync(MessageDataOptions payload)
        {
            // 批量写入
            TaskCompletionSource<Task> source = new TaskCompletionSource<Task>();
            _subject.OnNext(new BatchItemOptions<MessageDataOptions>
            {
                Body = payload,
                TaskSource = source
            });
            await source.Task;
        }

        public Task ShutdownAsync()
        {
            return Task.CompletedTask;
        }

        #region Private

        private async Task BatchWriteAsync(IList<BatchItemOptions<MessageDataOptions>> messages)
        {
            List<string> sqls = messages
                .Select(item => $"('{JsonConvert.SerializeObject(item.Body)}')").ToList();


            await using (SqlConnection conn = new SqlConnection(this.connection))
            {
                await conn.OpenAsync();
                var transaction = conn.BeginTransaction(System.Data.IsolationLevel.ReadUncommitted);
                try
                {
                    await using (SqlCommand command = new SqlCommand($"insert into {this.QueueTableName}(payload) values" + string.Join(',', sqls), conn, transaction))
                    {
                        await command.ExecuteNonQueryAsync();
                    }

                    await transaction.CommitAsync();

                    foreach (var item in messages)
                    {
                        item.TaskSource.TrySetResult(Task.CompletedTask);
                    }
                }
                catch (Exception ex)
                {
                    await transaction.RollbackAsync();
                    foreach (var item in messages)
                    {
                        item.TaskSource.TrySetException(ex);
                    }
                }
            }

        }

        #endregion
    }
}
